package database.impl;

import aof.Payload;
import aof.inter.IAddAof;
import config.ServerConfig;
import database.inter.IDB;
import database.singleDB.SingleDB;
import datastruct.dict.impl.JoConcurrentMap;
import datastruct.dict.inter.IDict;
import datastruct.dict.inter.IDictConsumer;
import datastruct.list.impl.JoList;
import datastruct.list.inter.IListConsumer;
import datastruct.sortedset.Element;
import datastruct.sortedset.JoSortedSet;
import exception.CustomException;
import jodis.conn.impl.*;
import jodis.conn.inter.IReply;
import parser.Parser;
import parser.ReadState;
import rdb.RDBConstant;
import rdb.RDBWriter;
import tcp.Server;
import tcp.utils.StreamUtils;

import java.io.*;
import java.util.HashMap;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReferenceArray;

public class MulIDB implements IDB, IAddAof {

    private AtomicReferenceArray<IDB> dbSet;
    private File aofFile;
    private int aofCurrentDb;
    private BlockingQueue<Payload> blockingQueue;
    private BufferedOutputStream aofOutputStream;
    private RDBWriter rdbWriter;

    public static final int DEFAULT_CHECK_KEY_NUM = 65535;

    public MulIDB() {
        dbSet = new AtomicReferenceArray<>(16);
        // 默认 0 号数据库
        aofCurrentDb = 0;

        // redis 默认有 16 个数据库
        for (int i=0;i<16;i++) {
            SingleDB db = SingleDB.makeDB();
            db.index = i;
            dbSet.set(i, db);
        }
        // 创建定时任务，执行键定期删除策略
        Timer timer = new Timer();
        TimerTask task = new TimerTask() {
            @Override
            public void run() {
                activeExpireCycle();
            }
        };
//         延迟 2s 后执行，每隔 2s 执行一次
        timer.schedule(task, 2000, 2000);


        // todo 初始化发布和订阅


        // todo 初始化主从复制

        // aof
        if (ServerConfig.getDefaultConfig().appendOnly) {
            aofFile = new File(Server.getServerConfig().appendFileName);
            blockingQueue = new LinkedBlockingQueue<>(1024);
            try {
                aofOutputStream = new BufferedOutputStream(new FileOutputStream(aofFile));
                for (int i=0;i<dbSet.length();i++) {
                    IDB idb = dbSet.get(i);
                    SingleDB dbImpl = (SingleDB) idb;
                    dbImpl.iAddAof = this;
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    // 定期删除策略
    public void activeExpireCycle() {
        for (int i=0;i<16;i++) {
            SingleDB selectDB = (SingleDB) dbSet.get(i);

            int keyNum = Math.min(selectDB.getTTLMap().len(), DEFAULT_CHECK_KEY_NUM);
            String[] randomKeys = selectDB.getTTLMap().randomDistinctKeys(keyNum);
            IDict ttlMap = selectDB.getTTLMap();
            for (int j=0;j<randomKeys.length;j++) {
                if (ttlMap.contains(randomKeys[j])) {
                    Long ttl = (Long) ttlMap.get(randomKeys[j]);
                    if (ttl <= System.currentTimeMillis()) {
                        selectDB.removeEntity(randomKeys[j]);
                        ttlMap.remove(randomKeys[j]);
                    }
                }
            }
        }
    }

    @Override
    public IReply exec(JoConnection clientConn, String[] cmdLine) {
        // 处理 select 命令
        if (cmdLine[0].equalsIgnoreCase("select")) {
            int selectDBIndex = Integer.parseInt(cmdLine[1]) % dbSet.length();
            clientConn.selectDB(selectDBIndex);
            return OkReply.makeOKReply();
        } else if (cmdLine[0].equalsIgnoreCase("command")) {
            return EmptyMultiBulkReply.makeEmptyMultiBulkReply();
        } else if (cmdLine[0].equalsIgnoreCase("ping")) {
            return PingReply.MakePingReply();
        }

        String cmdName = cmdLine[0];
        // 处理 save 命令
        if (cmdName.equalsIgnoreCase("save")) {
            return saveDB();
        }

        // 处理 config get xxx 命令
        if (cmdName.equalsIgnoreCase("config")) {
            if (cmdLine.length < 3) {
                return StandardErrReply.makeErrReply("illegal arguments for config get");
            }

            if (cmdLine[1].equalsIgnoreCase("get")) {
                if (cmdLine[2].equalsIgnoreCase("save")) {
                    return MultiBulkReply.makeMultiBulkReply(new String[]{cmdLine[2], "3600 1 300 100 60 10000"});
                } else if (cmdLine[2].equalsIgnoreCase("appendonly")) {
                    return MultiBulkReply.makeMultiBulkReply(new String[]{cmdLine[2], Server.getServerConfig().appendOnly ? "yes" : "no"});
                } else {
                    return EmptyMultiBulkReply.makeEmptyMultiBulkReply();
                }
            }
        }
        int dbIndex = clientConn.getDBIndex();
        IDB selectDB = dbSet.get(dbIndex);

        return selectDB.exec(clientConn, cmdLine);
    }

    //  生成 rdb 文件
    public IReply saveDB() {
        try {
            RDBWriter rdbWriter = new RDBWriter();
            rdbWriter.writeHeader();

            rdbWriter.writeAux("redis-ver", "6.0.10");
            rdbWriter.writeAux("redis-bits", "64");
            rdbWriter.writeAux("ctime", Integer.toString((int)(System.currentTimeMillis()/1000)));
            rdbWriter.writeAux("aof-preamble", "0");


            for (int i=0;i<16;i++) {
                SingleDB selectDB = (SingleDB) dbSet.get(i);
                int keyCount = selectDB.getKeyCount();
                int ttlCount = selectDB.getTTLCount();

                if (keyCount == 0 && ttlCount == 0) {
                    continue;
                }

                rdbWriter.writeDBHeader(selectDB.index, keyCount, ttlCount);

                selectDB.forEach(new IDictConsumer() {
                    @Override
                    public boolean consumer(String key, Object val) {
                        try {
                            System.out.printf("key: %s\n", key);
                            Long expiration = selectDB.getTTL(key);
                            int type = selectDB.getType(key);
                            switch (type) {
                                case RDBConstant.TYPE_STRING:
                                    if (selectDB.getEntity(key) == null) {
                                        System.out.println("@@@@@@@@@@ selectDB  getEntity(key) is null");
                                    }
                                    rdbWriter.writeStringObject(key, (String) selectDB.getEntity(key), expiration);
                                    break;
                                case RDBConstant.TYPE_LIST:
                                    JoList retList = (JoList) selectDB.getEntity(key);
                                    String[] vals = new String[retList.Len()];
                                    retList.forEach(false, new IListConsumer() {
                                        @Override
                                        public boolean consumer(int i, Object val) {
                                            vals[i] = (String) val;
                                            return true;
                                        }
                                    });
                                    rdbWriter.writeListObject(key, vals, expiration);
                                    break;
                                case RDBConstant.TYPE_SET:
                                    JoConcurrentMap dict = (JoConcurrentMap) selectDB.getEntity(key);
                                    rdbWriter.writeSetObject(key, dict.keys(), expiration);
                                    break;
                                case RDBConstant.TYPE_ZSET:
                                    JoSortedSet sortedSet = (JoSortedSet) selectDB.getEntity(key);
                                    Element[] elements = sortedSet.range(0, sortedSet.Len(), false);
                                    String[] members = new String[sortedSet.Len()];
                                    double[] scores = new double[sortedSet.Len()];
                                    for (int j = 0; j < elements.length; j++) {
                                        members[j] = elements[j].member;
                                        scores[j] = elements[j].score;
                                    }
                                    rdbWriter.writeZSetObject(key, members, scores, expiration);
                                    break;
                                case RDBConstant.TYPE_HASH:
                                    JoConcurrentMap hashTable = (JoConcurrentMap) selectDB.getEntity(key);
                                    HashMap<String, String> ht = new HashMap<>();
                                    hashTable.forEach(new IDictConsumer() {
                                        @Override
                                        public boolean consumer(String key, Object val) {
                                            ht.put(key, (String) val);
                                            System.out.printf("key: %s, val: %s\n", key, val);
                                            return true;
                                        }
                                    });
                                    rdbWriter.writeHashMapObject(key, ht, expiration);
                            }
                        } catch (IOException e) {
                            e.printStackTrace();
                        }

                        return true;
                    }
                });
                rdbWriter.flush();
            }
            // eof 和 检验和
            rdbWriter.writeEnd();
            rdbWriter.closeWriter();
        } catch (IOException e) {
            e.printStackTrace();
            return StandardErrReply.makeErrReply(e.getMessage());
        }
        return OkReply.makeOKReply();
    }

    public void loadAof() {
        if (aofFile == null || !aofFile.exists()) {
            return;
        }

        ByteArrayOutputStream byteStream = new ByteArrayOutputStream();
        ReadState state = ReadState.getDefaultInstance();
        FileInputStream fin = null;
        byte[] bs;
        try {
            fin = new FileInputStream(aofFile);
            while (true) {
                bs = StreamUtils.readStream(byteStream, fin, StreamUtils.readWait(), state);
                Parser.parse(bs, state);
                if (state.msgType == Parser.ARRAY) {
                    this.exec(JoConnection.fakeConn(0), state.resultArr);
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        } catch (CustomException ignore) {

        } finally {
            try {
                if (fin != null) {
                    fin.close();
                }
            } catch (IOException ee) {
                ee.printStackTrace();
            }
        }
    }

    @Override
    public void addAof(int dbIndex, String[] cmdLine) {
        Payload payload = new Payload(dbIndex, cmdLine);
        if (!blockingQueue.offer(payload)) {
            System.out.println("blocking queue 已满");
        }
    }

    public void handleAof() {
        if (!Server.getServerConfig().appendOnly) {
            return;
        }
        new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    while (true) {
                        Payload payload = blockingQueue.poll(2, TimeUnit.SECONDS);
                        if (payload != null) {
                            if (payload.dbIndex != aofCurrentDb) {
                                MultiBulkReply data = MultiBulkReply.makeMultiBulkReply(new String[] {
                                        "select", Integer.toString(payload.dbIndex)
                                });
                                aofOutputStream.write(data.toBytes());
                            }
                            aofOutputStream.write(new MultiBulkReply(payload.cmdLine).toBytes());
                            aofOutputStream.flush();
                        }
                    }
                } catch (InterruptedException | IOException e) {
                    e.printStackTrace();
                }
            }
        }).start();

    }
}
